import json

import requests
from ray.dashboard.modules.job.common import JobStatus
from ray.util.state import summarize_tasks, summarize_actors, summarize_objects
from ray.util.state import list_nodes, list_placement_groups, list_objects, list_actors, list_tasks, list_logs, \
    list_jobs, list_runtime_envs
from ray.util.state import get_task, get_node, get_log
import ray
import time


class RayManager:
    def __init__(self):
        pass

    @staticmethod
    def summarize_tasks():
        try:
            ret = summarize_tasks()
        except Exception as e:
            ret = str(e)
        return ret

    @staticmethod
    def summarize_actors():
        try:
            ret = summarize_actors()
        except Exception as e:
            ret = str(e)
        return ret

    @staticmethod
    def summarize_objects():
        try:
            ret = summarize_objects()
        except Exception as e:
            ret = str(e)
        return ret



    @staticmethod
    def list_placement_groups():
        try:
            ret = list_placement_groups(detail=True)
        except Exception as e:
            ret = str(e)
        return ret

    @staticmethod
    def list_objects():
        try:
            ret = list_objects(detail=True)
        except Exception as e:
            ret = str(e)
        return ret

    @staticmethod
    def list_nodes():
        try:
            ret = []
            node_list = list_nodes(detail=True)
            master_ip = ""
            for node in node_list:
                if node["is_head_node"]:
                    master_ip = node["node_ip"]
                    break
            request_addr = "http://"+master_ip + ":8265/nodes?view=summary"
            nodes = json.loads(requests.get(request_addr).text)["data"]["summary"]
            for node in nodes:
                node_id = node["cmdline"][7].split("=")[1]
                node_ip = node["ip"]
                node_name = node["hostname"]
                is_head_node = True if node["ip"] == master_ip else False
                cpu_percent = node["cpu"]
                cpu_phy = node["cpus"][1]
                cpu_logic = node["cpus"][0]
                disk = node["disk"]
                gpus = node["gpus"]
                memory_total = node["mem"][0]
                memory_free = node["mem"][1]
                memory_percent = node["mem"][2]
                status = node["status"]
                ret.append({"node_id": node_id,
                            "node_ip": node_ip,
                            "node_name": node_name,
                            "is_head_node": is_head_node,
                            "cpu_percent": cpu_percent,
                            "cpu_phy": cpu_phy,
                            "cpu_logic": cpu_logic,
                            "disk": disk,
                            "gpus": gpus,
                            "memory_total": memory_total,
                            "memory_free": memory_free,
                            "memory_percent": memory_percent,
                            "status": status})
            return {"error_num": 0, "msg": "获取成功", "data": ret}
        except Exception as e:
            ret = []
            return {"error_num": 1, "msg": "获取失败", "data": ret}

    @staticmethod
    def list_actors(condition):
        try:
            ret = []
            actor_list = list_actors(filters=condition, detail=True)
            for actor in actor_list:
                ret.append({"actor_id": actor["actor_id"],
                            "class_name": actor["class_name"],
                            "node_id": actor["node_id"],
                            "state": actor["state"]})
            return {"error_num": 0, "msg": "获取成功", "data": ret}
        except Exception as e:
            ret = []
            return {"error_num": 1, "msg": "获取失败", "data": ret}

    @staticmethod
    def list_tasks(condition):
        try:
            ret = []
            task_list = list_tasks(filters=condition)
            for task in task_list:
                ret.append({"task_id": task["task_id"],
                            "task_name": task["name"],
                            "state": task["state"],
                            "func_or_class_name": task["func_or_class_name"],
                            "node_id": task["node_id"],
                            "actor_id": task["actor_id"],
                            "worker_id": task["worker_id"],
                            "type": task["type"]})
            return {"error_num": 0, "msg": "获取成功", "data": ret}
        except Exception as e:
            ret = []
            return {"error_num": 1, "msg": "获取失败", "data": ret}

    @staticmethod
    def list_jobs(condition):
        try:
            ret = []
            job_list = list_jobs(filters=condition)
            for job in job_list:
                ret.append({"job_id": job["job_id"],
                            "status": job["status"].name,
                            "driver_info": {"id": job["driver_info"].id,
                                            "node_ip_address": job["driver_info"].node_ip_address,
                                            "pid": job["driver_info"].pid},
                            })
            return {"error_num": 0, "msg": "获取成功", "data": ret}
        except Exception as e:
            ret = []
            return {"error_num": 1, "msg": "获取失败", "data": ret}


if __name__ == "__main__":
    ray.init(address='ray://192.168.1.167:10001')
    @ray.remote
    class myactor1:
        def __init__(self):
            pass

        def ok(self):
            time.sleep(1)
            return 0


    # Create 2 actors
    actors = [myactor1.remote() for _ in range(2)]
    results = ray.get([c.ok.remote() for c in actors])
    print(results)
    # jobs = RayManager.list_jobs([])
    # tasks = RayManager.list_tasks([])
    # print(jobs)
    # print(tasks)

    # print(summarize_tasks())
    # pass